package com.example.xd.javatest.rxjava_test;


import android.util.Log;

import java.util.ArrayList;
import java.util.concurrent.TimeUnit;

import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.ObservableSource;
import io.reactivex.Observer;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.BiFunction;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;
import io.reactivex.schedulers.Schedulers;

/**
 * @auther XuDan on 2020/6/24.
 * 创建被观察者 创建观察者 订阅
 */
public class RxJavaTest {
    public static final String TAG = "xd---";

    /**
     * 基础版
     */
    public static void tset1() {
        Observable<String> observable = Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> e) throws Exception {
                e.onNext("1");
                e.onNext("2");
                e.onNext("3");
                e.onComplete();
            }
        });

        Observer observer = new Observer<String>() {
            @Override
            public void onSubscribe(Disposable d) {//最先调用的

            }

            @Override
            public void onNext(String value) {
                Log.i(TAG, value);
            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onComplete() {


            }
        };
        observable.subscribe(observer);
    }

    /**
     * Consumer 只接收onNext
     */
    public static void test2() {
        Observable.create(new ObservableOnSubscribe<String>() {

            @Override
            public void subscribe(ObservableEmitter<String> e) throws Exception {
                e.onNext("nihoa1");
                e.onNext("nihoa2");
                e.onComplete();
            }
        }).subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                Log.i(TAG, s);
            }
        });
    }

    public static void test3() {
        Observable.create(new ObservableOnSubscribe<Integer>() {

            @Override
            public void subscribe(ObservableEmitter<Integer> e) throws Exception {
                e.onNext(1);
                e.onNext(2);
                e.onNext(3);
                e.onComplete();
                e.onNext(4);
            }
        }).subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {

            }

            @Override
            public void onNext(Integer value) {
                Log.i(TAG, value.toString());
            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onComplete() {

            }
        });
    }

    /**
     * 线程切换,内置了线程池来管理,效率高
     */
    public static void test4() {
        Observable.create(new ObservableOnSubscribe<String>() {

            @Override
            public void subscribe(ObservableEmitter<String> e) throws Exception {
                e.onNext("111");
                e.onNext("111");
                e.onNext("111");
                Log.i(TAG, Thread.currentThread().getName());
            }
        }).subscribeOn(Schedulers.newThread())//指定上游发送事件的线程
                .observeOn(AndroidSchedulers.mainThread())//指定下游处理事件的线程
                .subscribe(new Consumer<String>() {
                    @Override
                    public void accept(String s) throws Exception {
                        Log.i(TAG, s);
                        Log.i(TAG, Thread.currentThread().getName());

                    }
                });
    }

    /**
     * 嵌套
     */
    public static void test5() {
        Observable.create(new ObservableOnSubscribe<Integer>() {

            @Override
            public void subscribe(ObservableEmitter<Integer> e) throws Exception {
                e.onNext(1);
                e.onNext(2);
                e.onNext(3);
            }
        }).flatMap(new Function<Integer, ObservableSource<String>>() {
            @Override
            public ObservableSource<String> apply(Integer integer) throws Exception {
                ArrayList<String> strings = new ArrayList<>();
                for (int i = 0; i < 3; i++) {
                    strings.add("i am value" + integer);
                }
                return Observable.fromIterable(strings).delay(10, TimeUnit.MILLISECONDS);
            }
        }).subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                Log.i(TAG, s);
            }
        });
    }

    /**
     * zip 将多个网络请求的数据合并转化以另一种数据
     * 参数类型要一一对应,否则会报参数类型异常
     */
    public static void test6() {
        Observable o1 = Observable.create(new ObservableOnSubscribe<String>() {

            @Override
            public void subscribe(ObservableEmitter<String> e) throws Exception {
                e.onNext("aa");
                e.onNext("bb");
                e.onNext("cc");
            }
        }).observeOn(Schedulers.newThread());

        Observable o2 = Observable.create(new ObservableOnSubscribe<Integer>() {

            @Override
            public void subscribe(ObservableEmitter<Integer> e) throws Exception {
                e.onNext(11);
                e.onNext(22);
                e.onNext(33);
                e.onNext(44);
            }
        }).observeOn(Schedulers.newThread());

        Observable.zip(o1, o2, new BiFunction<String, Integer, String>() {
            @Override
            public String apply(String s, Integer integer) throws Exception {
                return integer + "***" + s;
            }
        }).subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                Log.i(TAG, s);
            }
        });
    }

    /**
     * 背压缓存`区大小测试
     */
    public static void test7() {
    }
}
